//
//  tdl/mem/rblock_queue.h
//  Created by tommy on 13/04/24
//

#ifndef tdl_mem_rblock_queue_h
#define tdl_mem_rblock_queue_h
#include "sys/thread.h"
#include "mem/nolock_queue.h"
using namespace tdl;

namespace tdl {
///////////////
template <typename Tq> class rblock_queue_base
{
public:
	typedef typename Tq::size_type		size_type;

protected:
	Tq				m_nlq;
	tdl::thcond		m_cond;

public:
	rblock_queue_base() 
	{}

	virtual ~rblock_queue_base()
	{}
};

template <typename __element, template <typename> class __allocator = tdl::new_allocator>
class rblock_queue : public rblock_queue_base<tdl::nolock_queue<__element, __allocator> >
{
public:
	typedef rblock_queue_base<tdl::nolock_queue<__element, __allocator> > base;

	typedef typename base::size_type		size_type;

public:
	void initialize(size_type elm_num, key_t shm_key = -1) throw(alloc_ex)
	{
		base::m_nlq.initialize(elm_num, shm_key);
	}

	bool push(const __element& elm)
	{
		auto_locker<mutex_lock> al(base::m_cond.get_mutex());

		bool r = base::m_nlq.push_back(elm);
		if(r)
			base::m_cond.wakeup_one();

		return r;
	}

	void pop(__element& elm)
	{
		auto_locker<mutex_lock> al(base::m_cond.get_mutex());

		while(1)
		{
			if(base::m_nlq.pop_front(elm))
				return;

			base::m_cond.wait();
		}
	}
};

template <template <typename> class __allocator>
class rblock_queue<void, __allocator> : public rblock_queue_base<tdl::nolock_queue<void, __allocator> >
{
public:
	typedef rblock_queue_base<tdl::nolock_queue<void, __allocator> > base;

	typedef typename base::size_type		size_type;

public:
	void initialize(size_type qsize, key_t shm_key = -1) throw(alloc_ex)
	{
		base::m_nlq.initialize(qsize, shm_key);
	}

	template <typename T>
	bool push(const T& elm, size_type elm_len)
	{
		auto_locker<mutex_lock> al(base::m_cond.get_mutex());

		bool r = base::m_nlq.push_back(elm, elm_len);
		if(r)
			base::m_cond.wakeup_one();

		return r;
	}

	bool push(const char* pelm, size_type elm_len)
	{
		auto_locker<mutex_lock> al(base::m_cond.get_mutex());

		bool r = base::m_nlq.push_back(pelm, elm_len);
		if(r)
			base::m_cond.wakeup_one();

		return r;
	}

	template <typename T>
	void pop(T& elm, size_type& elm_len)
	{
		auto_locker<mutex_lock> al(base::m_cond.get_mutex());

		while(! base::m_nlq.pop_front(elm, elm_len))
			base::m_cond.wait();
	}

	void pop(char* pelm, size_type& elm_len)
	{
		auto_locker<mutex_lock> al(base::m_cond.get_mutex());

		while(! base::m_nlq.pop_front(pelm, elm_len))
			base::m_cond.wait();
	}
};

/////////////
template <template <typename> class __allocator = tdl::new_allocator>
class rblock_queue_ex : public rblock_queue<void, __allocator> {};

///////////////
}	// end of namespace
#endif
